//package com.xqb.server;
//
//import cn.hutool.core.util.StrUtil;
//import com.alibaba.otter.canal.client.CanalConnector;
//import com.alibaba.otter.canal.client.CanalConnectors;
//import com.alibaba.otter.canal.protocol.CanalEntry;
//import com.alibaba.otter.canal.protocol.Message;
//import com.google.protobuf.InvalidProtocolBufferException;
//import com.xqb.common.constants.MQConstants;
//import com.xqb.modules.domain.pojo.es.EsCourse;
//import com.xqb.modules.repository.EsCourseRepository;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.amqp.rabbit.core.RabbitTemplate;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.PostConstruct;
//import java.lang.reflect.Field;
//import java.net.InetSocketAddress;
//import java.text.ParseException;
//import java.text.SimpleDateFormat;
//import java.util.Date;
//import java.util.List;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Executors;
//
//import static com.xqb.common.constants.TableConstants.COURSE;
//import static com.xqb.common.constants.TableConstants.COURSE_INFO;
//
///**
// * @Author: JinChen Zhu
// * @CreateTime: 2024-10-29  12:03
// */
//@Component
//@Slf4j
////@ConfigurationProperties(prefix = "canal.config")
//public class CanalServer {
//    //Canal服务地址
//    @Value("${canal.config.server-address}")
//    private String serverAddress;
//    //Canal Server 服务端口号
//    @Value("${canal.config.port}")
//    private Integer port;
//    //目的地，其实Canal Service内部有一个队列,和配置文件中一致即可，参考【修改instance.properties】图中
//    @Value("${canal.config.destination}")
//    private String destination;
//    //用户名和密码，但是目前不支持，只能为空
//    @Value("${canal.config.username}")
//    private String username;
//    //用户名和密码，但是目前不支持，只能为空
//    @Value("${canal.config.password}")
//    private String password;
//    @Autowired
//    private EsCourseRepository esCourseRepository;
//    @Autowired
//    private RabbitTemplate rabbitTemplate;
//
//
//    @PostConstruct
//    public void init() throws Exception {
//        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(serverAddress, port), destination, username, password);
//        canalConnector.connect();
//        //订阅所有消息
////        canalConnector.subscribe(".*\\\\..*");
//        canalConnector.subscribe("db_xqb.*"); //只订阅test1数据库下的所有表
//        //恢复到之前同步的那个位置
//        canalConnector.rollback();
//        ExecutorService executor = Executors.newSingleThreadExecutor();
//        executor.submit(() -> {
//            while (true) {
//                long batchId = 0;
//                try {
//                    // 获取指定数量的数据，但是不做确认标记，下一次取还会取到这些信息。注：不会阻塞，若不够100，则有多少返回多少
//                    Message message = canalConnector.getWithoutAck(100);
//                    // 获取消息id
//                    batchId = message.getId();
//                    if (batchId != -1) {
//                        log.info("msgId -> " + batchId);
//                        // 确保 message.getEntries() 不是 null
//                        if (message.getEntries() != null) {
//                            //打印信息
//                            printEnity(message.getEntries());
//                        }
//                        // 提交确认
//                        canalConnector.ack(batchId);
//                    }
//                } catch (Exception e) {
//                    log.error("Error processing message: " + e.getMessage());
//                    // 处理失败，回滚数据
//                    canalConnector.rollback(batchId);
//                }
//            }
//        });
//
//
//    }
//
//    private void printEnity(List<CanalEntry.Entry> entries) {
//        for (CanalEntry.Entry entry : entries) {
//            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
//                continue;
//            }
//            try {
//                // 序列化数据
//                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
//                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
//                    String tableName = entry.getHeader().getTableName();
//                    CanalEntry.EventType eventType = rowChange.getEventType();
//                    // 表名
//                    log.info("表名：" + tableName);
//                    if (COURSE.equals(tableName) || COURSE_INFO.equals(tableName)) {                        //获取操作类型
//                        String id;
//                        List<CanalEntry.Column> afterColumnsList;
//                        switch (eventType) {
//                            //如果希望监听多种事件，可以手动增加case
//                            case INSERT:
//                            case UPDATE:
//                                afterColumnsList = rowData.getAfterColumnsList();
//                                if (COURSE.equals(tableName)) {
//                                    id = afterColumnsList.get(0).getValue();
//                                } else {
//                                    id = afterColumnsList.get(1).getValue();
//                                }
//                                log.info("准备发送更新id消息：" + id);
//                                //发送给rabbitmq
//                                rabbitTemplate.convertAndSend(MQConstants.COURSE_INSERT_QUEUE, id);
//                                break;
//                            case DELETE:
//                                List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
//                                id = beforeColumnsList.get(0).getValue();
//                                log.info("准备发送删除id消息：" + id);
//                                //发送给rabbitmq
//                                rabbitTemplate.convertAndSend(MQConstants.COURSE_DELETE_QUEUE, id);
//                                break;
//                            default:
//                        }
//                    }
//                }
//            } catch (InvalidProtocolBufferException e) {
//                log.error("解析异常", e);
//            }
//        }
//    }
//
//    public static EsCourse convertToEsCourse(List<CanalEntry.Column> columns) {
//        EsCourse esCourse = new EsCourse();
//        for (CanalEntry.Column c : columns) {
//            // 工具类转小驼峰
//            String name = StrUtil.toCamelCase(c.getName());
//            try {
//                // 尝试获取EsCourse中对应名称的字段
//                Field esCourseField = esCourse.getClass().getDeclaredField(name);
//
//                // 确保可以访问私有字段
//                esCourseField.setAccessible(true);
//
//                // 假设CanalEntry.Column有一个getValue()方法来获取字段值
//                // 并且这个值可以直接设置到EsCourse的对应字段中（类型匹配）
//                Object value = c.getValue(); // 你可能需要根据实际情况调整这部分代码
//                //检查字段类型，转成对应类型
//                if (esCourseField.getType().equals(value.getClass())) {
//                    esCourseField.set(esCourse, value);
//                } else if (esCourseField.getType().equals(Integer.class)) {
//                    esCourseField.set(esCourse, Integer.parseInt(value.toString()));
//                } else if (esCourseField.getType().equals(Double.class)) {
//                    esCourseField.set(esCourse, Double.parseDouble(value.toString()));
//                } else if (esCourseField.getType().equals(Float.class)) {
//                    esCourseField.set(esCourse, Float.parseFloat(value.toString()));
//                } else if (esCourseField.getType().equals(Long.class)) {
//                    esCourseField.set(esCourse, Long.parseLong(value.toString()));
//                } else if (esCourseField.getType().equals(Boolean.class)) {
//                    esCourseField.set(esCourse, Boolean.parseBoolean(value.toString()));
//                } else if (esCourseField.getType().equals(Date.class)) {
//                    //2022-07-23 16:46:33
//                    esCourseField.set(esCourse, new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(value.toString()));
//                } else {
//                    // 如果没有匹配的类型，则忽略这个字段
//                    log.warn("No matching type in EsCourse for {}", name);
//                }
//            } catch (NoSuchFieldException e) {
//                // 如果EsCourse中没有这个字段，则忽略它
////                log.warn("No matching field in EsCourse for {}", name);
//            } catch (IllegalAccessException e) {
//                // 处理字段访问异常（通常不应该发生，因为我们设置了setAccessible(true)）
//                log.error("Error accessing field", e);
//            } catch (ParseException e) {
//                log.error("Error parsing date", e);
//            }
//        }
//
////        log.info(esCourse.toString());
//        return esCourse;
//    }
//}
